package com.ksyun.campus.dataserver.services.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ksyun.campus.dataserver.domain.DataServer;
import com.ksyun.campus.dataserver.exception.OperationFailedException;
import com.ksyun.campus.dataserver.services.RegistService;
import com.ksyun.campus.dataserver.util.NetworkUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

@Service
@Slf4j
public class RegistServiceImpl implements RegistService, ApplicationRunner {

    @Value("${server.port}")
    private int serverPort;  // 服务端口号

    @Value("${az.rack}")
    private String rack;  // 服务所在机架

    @Value("${az.zone}")
    private String zone;  // 服务所在区域

    @Value("${storagePath}")
    private String STORAGE_PATH;  // 数据存储路径

    @Value("${capacityBytes}")
    private Integer initialCapacityBytes;  // 配置文件中的初始容量

    private Integer CAPACITY_BYTES;  // 当前容量

    private Integer usedBytes = 0;  // 初始已用容量为0

    private Integer fileTotal = 0;  // 初始文件数量为0

    private ZooKeeper zooKeeper;  // ZooKeeper 客户端实例

    @Value("${zk.address}")
    private String ZK_ADDRESS;  // ZooKeeper 地址

    @Value("${zk.rootPath}")
    private String ZK_ROOT_PATH;  // ZooKeeper 根路径

    private String serverPath;  // 当前服务在 ZooKeeper 中的注册路径

    private final ObjectMapper objectMapper = new ObjectMapper();  // JSON 序列化工具

    @Override
    public void run(ApplicationArguments args) throws Exception {
        CAPACITY_BYTES = initialCapacityBytes;  // 初始化容量为配置文件中的值
        registToCenter();  // 注册服务到中心
    }

    @Override
    public void registToCenter() {
        try {
            // 初始化 ZooKeeper 客户端
            zooKeeper = new ZooKeeper(ZK_ADDRESS, 80000, event -> {
            });


            // 确保 ZooKeeper 根路径存在，如果不存在则创建
            ensurePathExists(ZK_ROOT_PATH);

            String localIp = NetworkUtils.getLocalIp();  // 获取本地 IP 地址
            usedBytes = calculateUsedBytes();  // 计算已使用的存储容量
            fileTotal = calculateFileTotal();  // 计算文件总数

            // 扩容或缩容逻辑
            if (usedBytes > 0.8 * CAPACITY_BYTES) {
                int newCapacity = CAPACITY_BYTES * 2;
                if (newCapacity > CAPACITY_BYTES) {
                    CAPACITY_BYTES = newCapacity;
                    log.info("扩容至 " + CAPACITY_BYTES/1048576 + " MB");
                }
            } else if (usedBytes < 0.2 * CAPACITY_BYTES) {
                int newCapacity = Math.max(CAPACITY_BYTES / 2, initialCapacityBytes);
                if (newCapacity < CAPACITY_BYTES) {
                    CAPACITY_BYTES = newCapacity;
                    log.info("缩容至 " + CAPACITY_BYTES/1048576 + " MB");
                }
            }

            // 构建 DataServer 对象，用于保存当前节点的状态信息
            DataServer dataServer = DataServer.builder()
                    .rack(rack)
                    .zone(zone)
                    .ip(localIp)
                    .port(serverPort)
                    .used(usedBytes)
                    .capacity(CAPACITY_BYTES)
                    .fileTotal(fileTotal)
                    .build();

            // 将 DataServer 对象序列化为 JSON 字符串
            String serverInfoJson = objectMapper.writeValueAsString(dataServer);

            // 在 ZooKeeper 中创建临时节点，并保存节点信息
            serverPath = zooKeeper.create(ZK_ROOT_PATH + "/server-",
                    serverInfoJson.getBytes(StandardCharsets.UTF_8),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL);

            log.info("注册 DataServer at " + serverPath);

        } catch (Exception e) {
            log.error("注册到Zookeeper失败", e);
            throw new OperationFailedException("注册到Zookeeper失败: " + e.getMessage());
        }
    }

    @Override
    @Scheduled(initialDelay = 30000, fixedRate = 30000)
    public void reportStatus() {
        try {
            String localIp = NetworkUtils.getLocalIp();  // 获取本地 IP 地址
            usedBytes = calculateUsedBytes();  // 重新计算已使用的存储容量
            fileTotal = calculateFileTotal();  // 重新计算文件总数量

            // 扩容或缩容逻辑
            if (usedBytes > 0.8 * CAPACITY_BYTES) {
                int newCapacity = Math.min(CAPACITY_BYTES * 2, initialCapacityBytes * 8);
                if (newCapacity > CAPACITY_BYTES) {
                    CAPACITY_BYTES = newCapacity;
                    log.info("扩容至 " + CAPACITY_BYTES/1048576 + " MB");
                }
            } else if (usedBytes < 0.2 * CAPACITY_BYTES) {
                int newCapacity = Math.max(CAPACITY_BYTES / 2, initialCapacityBytes);
                if (newCapacity < CAPACITY_BYTES) {
                    CAPACITY_BYTES = newCapacity;
                    log.info("缩容至 " + CAPACITY_BYTES/1048576 + " MB");
                }
            }

            // 构建新的 DataServer 对象
            DataServer dataServer = DataServer.builder()
                    .rack(rack)
                    .zone(zone)
                    .ip(localIp)
                    .port(serverPort)
                    .used(usedBytes)
                    .capacity(CAPACITY_BYTES)
                    .fileTotal(fileTotal)
                    .build();

            // 将 DataServer 对象序列化为 JSON 字符串
            String updatedInfoJson = objectMapper.writeValueAsString(dataServer);

            // 更新 ZooKeeper 中的节点信息
            zooKeeper.setData(serverPath, updatedInfoJson.getBytes(StandardCharsets.UTF_8), -1);

//            log.info("DataServer 状态已更新: " + updatedInfoJson);

            // 将字节值转换为 MB 用于日志输出
            long usedMB = usedBytes / (1024 * 1024);
            long capacityMB = CAPACITY_BYTES / (1024 * 1024);

            // 在日志中输出以 MB 为单位的值
            log.info("DataServer 状态已更新: rack={}, zone={}, ip={}, port={}, used={} MB, capacity={} MB, fileTotal={}",
                    rack, zone, localIp, serverPort, usedMB, capacityMB, fileTotal);


        } catch (Exception e) {
            log.error("状态更新失败", e);
            throw new OperationFailedException("状态更新失败: " + e.getMessage());
        }
    }

    @Override
    public List<DataServer> getDslist() {
        List<DataServer> dsList = new ArrayList<>();
        try {
            List<String> children = zooKeeper.getChildren(ZK_ROOT_PATH, false);  // 获取 ZooKeeper 根路径下的所有子节点
            for (String child : children) {
                byte[] data = zooKeeper.getData(ZK_ROOT_PATH + "/" + child, false, null);
                String serverInfoJson = new String(data, StandardCharsets.UTF_8);
                log.info("Server Info: " + serverInfoJson);

                // 将 JSON 字符串反序列化为 DataServer 对象
                DataServer dataServer = objectMapper.readValue(serverInfoJson, DataServer.class);
                dsList.add(dataServer);
            }
        } catch (Exception e) {
            log.error("获取 DataServer 列表失败", e);
        }
        return dsList;
    }

    // 确保指定的 ZooKeeper 路径存在。如果路径不存在，则递归创建。
    private void ensurePathExists(String path) throws KeeperException, InterruptedException {
        String[] pathElements = path.split("/");
        StringBuilder currentPath = new StringBuilder();
        for (String element : pathElements) {
            if (!element.isEmpty()) {
                currentPath.append("/").append(element);
                Stat stat = zooKeeper.exists(currentPath.toString(), false);
                if (stat == null) {
                    zooKeeper.create(currentPath.toString(), null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                }
            }
        }
    }

    // 计算已使用的存储容量。
    private int calculateUsedBytes() {
        return calculateDirectorySize(new File(STORAGE_PATH));
    }

    // 递归计算目录及其子目录的总大小。
    private int calculateDirectorySize(File directory) {
        int size = 0;
        if (directory.exists()) {
            for (File file : directory.listFiles()) {
                if (file.isFile()) {
                    size += file.length();
                } else {
                    size += calculateDirectorySize(file); // 递归计算子目录大小
                }
            }
        }
        return size;
    }

    // 计算指定目录中的文件数量（不包括文件夹）。
    private int calculateFileTotal() {
        return calculateFileCount(new File(STORAGE_PATH));
    }

    // 递归计算指定目录中的文件数量。
    private int calculateFileCount(File directory) {
        int count = 0;
        if (directory.exists()) {
            for (File file : directory.listFiles()) {
                if (file.isFile()) {
                    count++;
                } else if (file.isDirectory()) {
                    count += calculateFileCount(file); // 递归计算子目录中的文件数量
                }
            }
        }
        return count;
    }

}
